package com.rz.spark.utils

import com.rz.spark.beans.Log
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 将原始日志文件转换成parquet文件格式
  * 使用自定义类的方式构建schema信息
  */
object Bzip2ParquetV2 {

  def main(args: Array[String]): Unit = {
    // 0 检验参数个数
    if (args.length !=3){
      println(
        """
          |com.rz.spark.utils.Bzip2Parquet
          |参数：
          | logInputPath
          | compressionCode <snappy, gzip, lzo>
          | resultOutputPath
        """.stripMargin
      )
      sys.exit()
    }

    // 1 接受程序参数
    val Array(logInputPath, compressionCode, resultOutputPath) =args

    // 2 创建sparkConf-》sparkContext
    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")
    // RDD 序列化到磁盘 worker与worker之间的数据传输
    sparkConf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    sparkConf.set("spark.sql.parquet.compression.codec", compressionCode)
    sparkConf.registerKryoClasses(Array(classOf[Log]))

    val sc = new SparkContext(sparkConf)

    val sqlContext = new SQLContext(sc)
    // 3 读取日志数据
    val rawdata = sc.textFile(logInputPath)
    // 4 根据业务需求对数据进行ETL
    val dataLog: RDD[Log] = rawdata.map(_.split(",", -1)).filter(_.length>=85).map(Log(_))
    // 5 将结果存储到本地磁盘
    val dataFrame = sqlContext.createDataFrame(dataLog)
    dataFrame.write.partitionBy("provincename","cityname").parquet(resultOutputPath)
    // 6 关闭sc
    sc.stop()
  }
}
